package quasar;
import quasar_framework.Common;
import quasar_framework.KeyToIDMapper;

import il.technion.ewolf.kbr.Key;
import il.technion.ewolf.kbr.KeybasedRouting;
import il.technion.ewolf.kbr.MessageHandler;
import il.technion.ewolf.kbr.Node;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import quasar_framework.Message;
import quasar_framework.MessageType;
import quasar_framework.NieghborWalkaroundReply;
import quasar_framework.PublishMessageData;
import quasar_framework.RouteLookupEntry;
import quasar_framework.SubscribeMessageData;

import quasar_framework.MyBloomFilter;
import tweeter.NetworkEntity;

/**
 * This is the main class of the Quasar algorithm. Each Station instance implements Quasar
 * Publish-Subscribe algorithm according to the network traffic. Each station contains a
 * state, and is invoked using callback provided for the KeyBasedRouting class.
 */
public class Station extends NetworkEntity
{
	/** The KeyBasedRouting instance is used in order to connect and communicate with the network */
	public KeybasedRouting kbr;
	/** The routing table which contains information about requested subscriptions around the station */
	public ArrayList<HashMap<Key, MyBloomFilter>> routingTable;
	/** An extension of the routing table, which also stores the stations which can be reached using a
	 *  particular instance in the routing table */
	public ArrayList<HashMap<Key, HashSet<Integer>>> routingTableAux;
	/** The "recognition" ring of the station, the radius defines the ring of known stations for this
	 *  station in regard of routing. All the stations whose distance is farther than the radius, will be
	 *  considered in the outer ring. */
	public int radius;
	/** Defines how many max publications should be created each time for the "Random walk" */
	public int alpha;
	/** A list which contains the neighbors of the station */
	public List<Node> realNeighbours;
	/** The subscriptions that the current station requires */
	public HashSet<Integer> subscriptions;
	/** Logging data which contains the already received subscription messages, to avoid spamming the network */
	public HashSet<Integer> receivedSubscriptionMessages;
	/** Logging data which contains the already received publication messages, to avoid spamming the network */
	public HashSet<Integer> receivedPublicationMessages;
	
	public ArrayList<HashSet<Node>> radiusNeighbors;
	public HashMap<Key, Key> originators;
	public Integer expecting;
	public final Object sharedLock = new Object(); 
	public boolean useRouteTable;
	
	public HashMap<Key, Integer> mapz; 
	/**
	 * A basic constructor. Variables such as the alpha and the radius are retrieved using general information
	 * found in the JointDefinitions class.
	 * 
	 * @param ID - The ID the station will use. For publishing stations, that the ID of the publication.
	 * @param kbr - The KeyBasedRouting instance to be used for this instance.
	 */
	public Station(int ID, KeybasedRouting kbr, boolean useRouteTable)
	{
		super(ID);
		this.radius = Common.radius;
		this.alpha = Common.alpha;
		this.kbr = kbr;
		this.useRouteTable = useRouteTable;
		this.routingTable = new ArrayList<HashMap<Key, MyBloomFilter>>();
		this.routingTableAux = new ArrayList<HashMap<Key, HashSet<Integer>>>();
		this.realNeighbours = new ArrayList<Node>();
		this.subscriptions = new HashSet<Integer>();
		this.receivedSubscriptionMessages = new HashSet<Integer>();
		this.receivedPublicationMessages = new HashSet<Integer>();
		this.mapz = KeyToIDMapper.mapper;
		for (int i=0 ; i < radius ; i++) 
		{
			this.routingTable.add(new HashMap<Key, MyBloomFilter>());
			this.routingTableAux.add(new HashMap<Key, HashSet<Integer>>());
		}
		
		this.kbr.register(Common.subscriptionTag, new MessageHandler()
		{
			@Override
			public void onIncomingMessage(Node from, String tag, Serializable content)
			{
				if (!tag.equals(Common.subscriptionTag))
				{
					return;
				}
				handleSubscribeMessage(from, (Message)content);
			}

			@Override
			public Serializable onIncomingRequest(Node arg0, String arg1, Serializable arg2)
			{
				return null;
			}
		});
		
		this.kbr.register(Common.publicationTag, new MessageHandler()
		{
			@Override
			public void onIncomingMessage(Node from, String tag, Serializable content)
			{
				if (!tag.equals(Common.publicationTag))
				{
					return;
				}
				synchronized(Statistics.publishPacketTraffic)
				{
					Statistics.publishPacketTraffic++;
				}
				handlePublishMessage(from, (Message)content);
			}
			
			@Override
			public Serializable onIncomingRequest(Node arg0, String arg1, Serializable arg2)
			{
				return null;
			}
		});
		
		this.kbr.register(Common.symmetricP2PTag, new MessageHandler()
		{
			@Override
			public void onIncomingMessage(Node from, String tag, Serializable content)
			{
				if (!tag.equals(Common.symmetricP2PTag))
				{
					return;
				}
				if (updateRoutingTableWithNeighbour(from))
				{				
					// Sending an ACK back, if the neighbor is a new one!
					sendHiBack(from);
				}
			}
			
			@Override
			public Serializable onIncomingRequest(Node arg0, String arg1, Serializable arg2)
			{
				return null;
			}
		});

		mapz.put(kbr.getLocalNode().getKey(), super.ID);
		
		radiusNeighbors = new ArrayList<HashSet<Node>>();
		for (int index = 0 ; index < this.radius ; index++) radiusNeighbors.add(new HashSet<Node>());
		
		this.kbr.register(Common.neighborRequestTag, new MessageHandler()
		{
			@Override
			public void onIncomingMessage(Node from, String tag, Serializable content)
			{
				if (!tag.equals(Common.neighborRequestTag))
				{
					return;
				}
				neighborRequestHandling(from, (Integer)content);
			}
			
			@Override
			public Serializable onIncomingRequest(Node arg0, String arg1, Serializable arg2)
			{
				return null;
			}
		});
		
		this.kbr.register(Common.neighborReplyTag, new MessageHandler()
		{
			@Override
			public void onIncomingMessage(Node from, String tag, Serializable content)
			{
				if (!tag.equals(Common.neighborReplyTag))
				{
					return;
				}
				nieghborReplyHandling(from, (Message)content);
			}
			
			@Override
			public Serializable onIncomingRequest(Node arg0, String arg1, Serializable arg2)
			{
				return null;
			}
		});
		
		this.kbr.register(Common.mySubscriptionsTag, new MessageHandler()
		{
			@Override
			public void onIncomingMessage(Node from, String tag, Serializable content)
			{
				if (!tag.equals(Common.mySubscriptionsTag))
				{
					return;
				}
				routeUpdate(from, (Message)content);
			}
			
			@Override
			public Serializable onIncomingRequest(Node arg0, String arg1, Serializable arg2)
			{
				return null;
			}
		});
		
		 this.originators = new HashMap<Key, Key>();
	}
	
	public void sendAllYourSubscriptions()
	{
		System.out.println("Station " + super.ID + " is sending all its subscription data");
		
		for (int level = 0 ; level < this.radius ; level++)
		{
			HashSet<Node> toSend = this.radiusNeighbors.get(level);
			for (Node n : toSend)
			{
				Message m = new Message(MessageType.SUBSCRIBE, 0, 0);
				SubscribeMessageData md = new SubscribeMessageData(super.ID, this.subscriptions, Integer.toString(level));
				m.data = (Object)md;
				try
				{
					this.kbr.sendMessage(n, Common.mySubscriptionsTag, m);
					Thread.sleep(5);
				}
				catch (Exception e)
				{
					e.getStackTrace();
				}
			}
		}
		
		System.out.println("Station " + super.ID + " is done");
	}
	
	public void routeUpdate(Node from, Message m)
	{
		
		SubscribeMessageData messageData = (SubscribeMessageData)m.data;
		try
		{
			// Without thinking, adding the subscriptions to the matching level 
			synchronized(this.routingTable)
			{
				try
				{
					// Finding the originator of the message
					Key originatorKey = this.originators.get(from.getKey());
//					System.out.println("Station " + super.ID + " updating subscriptions from " + this.mapz.get(from.getKey()) + " (originator is " + this.mapz.get(originatorKey) + ")" + 
//							           " on level " + Integer.parseInt(messageData.payload) + ". Subs: " + messageData.subscriptions);
					
					// Adding all the subscriptions to the proper entry
					for (Integer subscription : messageData.subscriptions)
					{
						routingTable.get(Integer.parseInt(messageData.payload)).get(originatorKey).add(subscription);
					}
					routingTableAux.get(Integer.parseInt(messageData.payload)).get(originatorKey).add(messageData.subscriberID);
				}
				catch (Exception e)
				{
					System.out.println("Station " + super.ID + " encountered an error while updating the routing table with subscription " +
									   "information from station " + mapz.get(from.getKey()) +". Error is " + e.getLocalizedMessage());
					return;
				}
			}
		}
		catch (Exception e)
		{
			System.out.println("Station " + super.ID + " encountered an error while handling a received subscription. Error is " + e.getLocalizedMessage());
			return;
		}
	}
	
	public void neighborRequestHandling(Node from, Integer level)
	{
		Message m = new Message(null, 0, 0);
		NieghborWalkaroundReply data = new NieghborWalkaroundReply(this.realNeighbours, level);
		m.data = data;
		try
		{
			//System.out.println(super.ID + " sending neighbor to " + this.mapz.get(from.getKey())+ " for distance " +level + " reply with -> " + this.realNeighbours);
			this.kbr.sendMessage(from, Common.neighborReplyTag, m);
		} catch (IOException e)
		{
			e.printStackTrace();
		}
	}
	public void nieghborReplyHandling(Node from, Message m)
	{
		NieghborWalkaroundReply data = (NieghborWalkaroundReply)m.data;
		//System.out.println(super.ID + " got neighborReply from " + this.mapz.get(from.getKey()) +  " for distance " + data.distance + "contains -> " + data.neighbors);
//		data.neighbors.remove(this.kbr.getLocalNode());
		synchronized(this.radiusNeighbors) 
		{
			for (Node n : data.neighbors)
			{
				if (!this.radiusNeighbors.get(data.distance).contains(n))
				{
					this.radiusNeighbors.get(data.distance).add(n);
				}
			}
			
		}
		
		// Update the originators!
		synchronized(this.originators)
		{
			Key originatorKey = this.originators.get(from.getKey());
			// It's one of the 1st class neighbors, he's an originator!
			if (this.originators.containsKey(from.getKey()) && originatorKey.equals(from.getKey()))
			{
//				System.out.println(super.ID + " and " + this.mapz.get(from.getKey()) + " are neighbors");
				for (Node n : data.neighbors)
				{
					if (!this.originators.containsKey(n.getKey()))
					{
//						System.out.println("Adding " + this.mapz.get(originatorKey) + " <- " + this.mapz.get(n.getKey()));
						this.originators.put(n.getKey(), from.getKey());
					}
				}
			}
	
			// There's already an entry like this, but he's not an originator
			else if (this.originators.containsKey(from.getKey()))
			{		
//				System.out.println(super.ID + " and " + this.mapz.get(from.getKey()) + " are NOT neighbors, the originator is " + this.mapz.get(originatorKey));
				// We have to use the sender's originator as the originator for all that he brought
				for (Node n : data.neighbors)
				{		
					if (!this.originators.containsKey(n.getKey()))
					{
//						System.out.println("Adding " + this.mapz.get(originatorKey) + " <- " + this.mapz.get(n.getKey()));
						this.originators.put(n.getKey(), originatorKey);
					}
				}
			}
			
			// He's not here? WTF?
			else
			{
				System.out.println("*** HAAAAAAAAAAAAAAAAAAAAAAAA ***");
			}
		}
		
		synchronized(sharedLock)
		{
			this.expecting--;
			if (0 == expecting)
			{
			sharedLock.notify();
			}
		}
	}
	
	public void fillNeighbors(int level)
	{
		if (0 == level)
		{
			synchronized(this.originators)
			{
				radiusNeighbors.get(0).addAll(this.realNeighbours);
				for (Node neighbor : this.realNeighbours)
				{
					this.originators.put(neighbor.getKey(), neighbor.getKey());
				}
			}
		}
		else
		{			
			this.expecting = this.radiusNeighbors.get(level - 1).size();
			// Now we can iterate over what's left
			Iterator<Node> it = this.radiusNeighbors.get(level - 1).iterator();
			while (it.hasNext())
			{
				try
				{
					Node s = it.next();
					//System.out.println(super.ID + " is sending neighbor request to " + this.mapz.get(s.getKey()));
					this.kbr.sendMessage(s, Common.neighborRequestTag, new Integer(level));
					Thread.sleep(5);
				}
				catch (Exception e)
				{
					e.printStackTrace();
				}
			}
			
			synchronized(sharedLock)
			{
				try
				{
					while (0 != expecting)
					{
						sharedLock.wait();
					}
				}
				catch (InterruptedException e)
				{
					e.printStackTrace();
				}
			}
			
			
			// Cleanup, removing any nodes which are in a higher level
			for (int checkRadius = 0 ; checkRadius < level ; checkRadius++)
			{
				for (Node n : this.radiusNeighbors.get(checkRadius))
				{
					if (this.radiusNeighbors.get(level).contains(n))
					{
						this.radiusNeighbors.get(level).remove(n);
					}
				}
			}
			
			// In case I'm here as well
			this.radiusNeighbors.get(level).remove(this.kbr.getLocalNode());
		}
		
//		System.out.print(super.ID + " level " + level + " radius neighbors:");
//		for (Node n : this.radiusNeighbors.get(level))
//		{
//			System.out.print(" " + this.mapz.get(n.getKey()) + " ");
//		}
//		System.out.println();
//		System.out.println(super.ID + " originators:");
//		for (Key k : this.originators.keySet())
//		{
//			System.out.println(this.mapz.get(k) + " <- " + this.mapz.get(this.originators.get(k)));
//		}
	}
	
	/**
	 * The function is used in order to establish a symmetrical P2P network.
	 * The current station sends a confirmation to the sending station to establish that connection for
	 * both sides.
	 * 
	 * @param sendTo - The node to be addressed with the ACK. 
	 */
	private void sendHiBack(Node sendTo)
	{
		try
		{
			kbr.sendMessage(sendTo, Common.symmetricP2PTag, null);
		}
		catch (Exception e)
		{
			System.out.println("Station " + super.ID + " encountered an error while trying to send a P2P ACK. Error is " + e.getLocalizedMessage());
		}
	}

	/**
	 * The function updates the routing table with the information provided by the neighbor. Only the immediate
	 * neighbors send this information (The P2P message) and the routing table is built accordingly.
	 * 
	 * @param neighbour - The sending neighbor who should be inserted to the routing table
	 * @return true in case the routing table has been updated (the neighbor is new), otherwise false.
	 */
	private boolean updateRoutingTableWithNeighbour(Node neighbour)
	{
		synchronized(this.realNeighbours)
		{
			if (false == this.realNeighbours.contains(neighbour))
			{
				synchronized(this.routingTable)
				{
					for (int i = 0 ; i < radius ; i++)
					{
						routingTable.get(i).put(neighbour.getKey(), new MyBloomFilter(Common.bloomFilterSize));
						routingTableAux.get(i).put(neighbour.getKey(), new HashSet<Integer>());
					}
					realNeighbours.add(neighbour);
				}
				return true;
			}
			else
			{
				return false;
			}
		}
	}
	
	/**
	 * An internal function used for debugging purposes, it outputs the route of the message. In subscription messages,
	 * all the stations appear in the list, while in the publication messages only the originator and the recipients appear.
	 * 
	 * @param message - The routed message.
	 */
	private void outputMessageRoute(Message message)
	{
		String route = "Message ID " + message.ID + " Type " + message.type + " Route: " + message.alreadyReceived.toString();
		System.out.println(route);
	}
	
	private RouteLookupEntry findRouteEntry(Node from, Message message, PublishMessageData messageData)
	{
		RouteLookupEntry entry = new RouteLookupEntry(-1, null);
		
		for (int i = 0; (i < radius) && (-1 == entry.level); i++)
		{
			for (Key nodeKey : routingTable.get(i).keySet())
			{
				// Not sending back to the sender
				if (nodeKey.equals(from.getKey()))
				{
					continue;
				}
				
				// It the message is already sent to a node which is achievable by this bloom filter, continue
				
				boolean alreadyReceivedFlag = false;
				for (int alreadyReceivedStation : message.alreadyReceived)
				{
					if (routingTableAux.get(i).get(nodeKey).contains(alreadyReceivedStation))
					{
						alreadyReceivedFlag = true;
						break;
					}
				}
					
				if (alreadyReceivedFlag)
				{
					continue;
				}
						
				// Found a matching entry!
				if (routingTable.get(i).get(nodeKey).contains(messageData.groupID))
				{
					entry.neighborKey = nodeKey;
					entry.level = i;
					return entry;
				}
			}
		}
		
	return entry;
	}
	
	private void sendPublicationUsingRoutingTable(Node from, Message message, PublishMessageData messageData)
	{
		// Checking the routing table entries to see if any level matches
		// the publication request
		if (message.ttl > 0)
		{
			RouteLookupEntry entry = null;
			if (true == this.useRouteTable)
			{
				entry = findRouteEntry(from, message, messageData);
			}
			else
			{
				entry = new RouteLookupEntry(-1, null);
			}
			
			// There's no entry on the routing table which requires the
			// publication, so we'll send to a neighbor in random
			if (-1 == entry.level)
			{
				ArrayList<Node> toSend = fetchRandomRecievers(this.realNeighbours, 1, message.alreadyReceived);
				
				if (1 == toSend.size())
				{

					try
					{
						this.kbr.sendMessage(toSend.get(0), Common.publicationTag, message);
					}
					catch (Exception e)
					{
						System.out.println("Station " + super.ID + " encountered an error while trying to randomly send a publication. Error is " + e.getLocalizedMessage());
						return;
					}
				}
				else
				{
					outputMessageRoute(message);
				}
			}
				
			// Found a routing table entry which requires the publication
			else
			{
				Node nextHop = null;
				for (Node neighbourCandidate : realNeighbours)
				{
					if (neighbourCandidate.getKey().equals(entry.neighborKey))
					{
						nextHop = neighbourCandidate;
						break;
					}
				}
					
				if (null == nextHop)
				{
					System.out.println("Station " + super.ID + " encountered an error while trying to randomly send a publication. This is not expected at all!");
					return;
				}
					
				else
				{
	 				try
	 				{
	 					this.kbr.sendMessage(nextHop, Common.publicationTag, message);
	 				}
	 				catch (Exception e)
	 				{
	 					System.out.println("Station " + super.ID + " encountered an error while trying to route a publication using the routing table. Error is " + e.getLocalizedMessage());
	 					return;
	 				}
				}
			}
		}
		else
		{
			outputMessageRoute(message);
		}
	}
	
	/**
	 * The function is in charge of handling publication messages in the system. In case the publication is
	 * intended for it, it processes it, adds the necessary information, and routes it randomly according to
	 * Quasar. Otherwise, it either routes it according to the routing table, or to a random neighbor in case
	 * it is the originator of the message and it has been directed back to it.
	 * 
	 * @param from - The station from which the message is sent.
	 * @param message - The message sent.
	 */
	private void handlePublishMessage(Node from, Message message)
	{
		try
		{
			PublishMessageData messageData = (PublishMessageData)message.data;
			
			message.ttl--;
			message.hops++;
	
			// 1.
			// If the message is sent to the publisher and has more TTL, It is sent to a random neighbor
			if (super.ID == (messageData.publisherID))
			{
				if (message.ttl > 0)
				{			
					ArrayList<Node> toSend = fetchRandomRecievers(this.realNeighbours, 1, message.alreadyReceived);
					
					if (1 == toSend.size())
					{
						try
						{
							this.kbr.sendMessage(toSend.get(0), Common.publicationTag, message);
						}
					
						catch (Exception e)
						{
							System.out.println("Station " + super.ID + " encountered an error while trying to randomly send a publication. Error is " + e.getLocalizedMessage());
							return;
						}
					}
				}
				outputMessageRoute(message);
			}
			
			// 2.
			// If the station is a subscriber for this message and he didn't receive it already, he zeroes the TTL, and sends it to alpha neighbors randomly
			else if (this.subscriptions.contains(messageData.groupID))
			{
					
				// Just for tracking
				synchronized(this.receivedPublicationMessages)
				{
					// If the message hasn't been received already
					if (!this.receivedPublicationMessages.contains(message.ID))
					{
						synchronized(Statistics.deliveredPublishPackets)
						{
							Statistics.deliveredPublishPackets++;
						}
						this.receivedPublicationMessages.add(message.ID);
						// Adding it to the already received list of this message
						message.alreadyReceived.add(super.ID);
						System.out.println("Subscriber " + super.ID + " received the publication from " + messageData.publisherID);
					}
						
					// Already received the publication, working like a regular dot on the route, END OF THE LINE HERE!
					else
					{
						System.out.println("Subscriber " + super.ID + " ALREADY received the publication " + message.ID);
						sendPublicationUsingRoutingTable(from, message, messageData);
						return;
					}
				}
									
				// Continuing as planned, renewing the TTL and sending it to Alpha neighbors
				message.ttl = Common.publicationTTL;
//				System.out.println("Subscriber " + super.ID + "is routing message " +  message.ID + " to ALPHA potential subscribers");
				ArrayList<Node> toSend = fetchRandomRecievers(this.realNeighbours, Common.alpha, message.alreadyReceived);
				if (0 < toSend.size())
				{
					for (Node potentialReceiver : toSend)
					{
						try
						{
							this.kbr.sendMessage(potentialReceiver, Common.publicationTag, message);
						}
						catch (Exception e)
						{
							System.out.println("Station " + super.ID + " encountered an error while trying to randomly send a publication. Error is " + e.getLocalizedMessage());
						}
					}
				}
				outputMessageRoute(message);
			}
			
			// Routing the message regularly
			else
			{
				sendPublicationUsingRoutingTable(from, message, messageData);
			}
		}	
		catch (Exception e)
		{
			System.out.println("Station " + super.ID + " encountered an error while handling a received publication. Error is " + e.getLocalizedMessage());
			return;
		}
	}
	
	/**
	 * The function is charge of handling subscription messages according to the Quasar algorithm.
	 * Unless the subscription message has already been received by this station, it basically updates
	 * its routing table according to the subscription information in the message, and routes it
	 * to its neighbors who haven't received it yet.
	 * 
	 * @param from - The station from which the message is sent.
	 * @param message - The message sent.
	 */
	private void handleSubscribeMessage(Node from, Message message)
	{
		synchronized(Statistics.subscribePacketTraffic)
		{
			Statistics.subscribePacketTraffic++;
		}
		SubscribeMessageData messageData = (SubscribeMessageData)message.data;
		
		synchronized(this.receivedSubscriptionMessages)
		{
			// Checking if this message has already been received by the station,
			// if so, there's no need it processing and spamming it again
			if (this.receivedSubscriptionMessages.contains(message.ID))
			{
				outputMessageRoute(message);
				return;
			}
			else
			{
				this.receivedSubscriptionMessages.add(message.ID);
			}
		}
		
		// If message already received, it is dropped
		if (message.alreadyReceived.contains(super.ID))
		{
			outputMessageRoute(message);
			return;
		}
		
		message.alreadyReceived.add(super.ID);
		
		// Checking to which level the message matches, that way we'll know which level to
		// update in the routing table.
		// If the message was sent from "out of the station radius", the entry will be
		// added to the farthest radius level
		int level = Common.subscriptionTTL - message.ttl;
		if (level >= radius)
		{
			level = radius - 1;
		}
		
		message.ttl--;
		message.hops++;

		try
		{
			// Without thinking, adding the subscriptions to the matching level 
			synchronized(this.routingTable)
			{
				try
				{
					Iterator<Integer> subscriptionsIterator = messageData.subscriptions.iterator();
					while (subscriptionsIterator.hasNext())
					{
						routingTable.get(level).get(from.getKey()).add(subscriptionsIterator.next());
					}
					routingTableAux.get(level).get(from.getKey()).add(messageData.subscriberID);
				}
				catch (Exception e)
				{
					System.out.println("Station " + super.ID + " encountered an error while updating the routing table with subscription " +
									   "information from station " + mapz.get(from.getKey()) +". Error is " + e.getLocalizedMessage());
					return;
				}
			}
		}
		catch (Exception e)
		{
			System.out.println("Station " + super.ID + " encountered an error while handling a received subscription. Error is " + e.getLocalizedMessage());
			return;
		}
		
		if (0 < message.ttl)
		{
			try
			{
				// Sending the subscription message to all the neighbors
				for (Node neighbour : realNeighbours)
				{
					// Not sending to the sender, or anyone who already received it
					if ((mapz.get(neighbour.getKey()) == super.ID) ||
						(message.alreadyReceived.contains(mapz.get(neighbour.getKey()))))
					{
						continue;
					}
										
					this.kbr.sendMessage(neighbour, Common.subscriptionTag, message);
				}
			}
			catch (Exception e)
			{
				System.out.println("Station " + super.ID + " encountered an error while trying to forward a subscription to its neighbors. Error is " + e.getLocalizedMessage());
				return;
			}
			
		}
		else
		{ 
			outputMessageRoute(message);
		}
	}
	
	@Override
	/**
	 * The function originates a subscription message which the given ID to all of the
	 * neighbors.
	 * 
	 * @param messageID - The ID of the message which is going to be originated.
	 */
	public void subscribe(int messageID)
	{
		Message message = new Message(MessageType.SUBSCRIBE, this.radius, messageID);
		SubscribeMessageData messageData = new SubscribeMessageData(super.ID, this.subscriptions, null);
		message.data = messageData;
		message.alreadyReceived.add(super.ID);
		message.ttl = Common.subscriptionTTL;
		message.hops = 0;
		
		// Sending to all the stations' neighbors
		try
		{	
			// Send to all your neighbors
			for (Node neighbour : realNeighbours)
			{
				this.kbr.sendMessage(neighbour, Common.subscriptionTag, message);
			}
		}
		catch (Exception e)
		{
			System.out.println("Station " + super.ID + " encountered an error while trying to send its subscription information " +
		                       "to all of its neighbors. Error is " + e.getLocalizedMessage());
			return;
		}
		
	}
	
	@Override
	/**
	 * The function originates a publication message which the given ID to max alpha neighbors
	 * 
	 * @param messageID - The ID of the message which is going to be originated.
	 * @param messageContent - Any additional content required to be attached to the message.
	 */
	public void publish(int messageID, String messageContent)
	{
		Message message = new Message(MessageType.PUBLISH, this.radius, messageID);
		PublishMessageData messageData = new PublishMessageData(super.ID, super.ID, messageContent);
		message.data = messageData;
		message.alreadyReceived.add(super.ID);
		message.ttl = Common.publicationTTL;
		message.hops = 0;
		
		int leftTosend = this.alpha;
		HashSet<Integer> alreadySent = new HashSet<Integer>();
		boolean sentAtLeastOne = false;
		
		// Sending the publication to maximum alpha neighbors, to spread the word
		while (0 != leftTosend && alreadySent.size() != realNeighbours.size())
		{
			Random generator = new Random();
			
			int candidate = generator.nextInt(realNeighbours.size());
			
			while (alreadySent.contains(candidate))
			{
				candidate = generator.nextInt(realNeighbours.size());
			}
			
			try
			{
				this.kbr.sendMessage(realNeighbours.get(candidate), Common.publicationTag, message);
				alreadySent.add(candidate);
				leftTosend--;
				sentAtLeastOne = true;
			}
			catch (Exception e)
			{
				System.out.println("Station " + super.ID + " encountered an error while trying to randomly send its publication. Error is " + e.getLocalizedMessage());
			}
		}
		
		// If at least one publication has been sent, update the statistics
		if (sentAtLeastOne)
		{
			Statistics.addPublication(super.ID);
			synchronized(Statistics.sentPublications)
			{
				Statistics.sentPublications++;
			}
		}
	}
	
	/**
	 * An internal function which is used in order to retrieve a set of random receives, mainly
	 * for publication messages.
	 * 
	 * @param neighbors - A list of the potential neighbors that can receive the message.
	 * @param maxNumReceivers - The maximal number of neighbors to be chosen for sending.
	 * @param alreadyReceived - A list of neighbors (by IDs) who already recieved the message.
	 * @return
	 */
	private ArrayList<Node> fetchRandomRecievers(List<Node> neighbors, int maxNumReceivers, ArrayList<Integer> alreadyReceived)
	{
		HashSet<Node> alreadyTried = new HashSet<Node>();
		ArrayList<Node> receivers = new ArrayList<Node>();
		int runningIndex = 0;
		Random generator = new Random();
		
		while (runningIndex < maxNumReceivers &&
			   (neighbors.size() > (alreadyTried.size() + receivers.size())))
		{
			int candidate = generator.nextInt(neighbors.size());
			Node nodeCandidate = neighbors.get(candidate);
			
			// If we already checked it
			if (alreadyTried.contains(nodeCandidate))
			{
				continue;
			}
			
			// If it's already in the received list
			if (alreadyReceived.contains(nodeCandidate))
			{
				alreadyTried.add(nodeCandidate);
				continue;
			}
			
			// We can add him!
			receivers.add(nodeCandidate);
			alreadyTried.add(nodeCandidate);
			runningIndex++;
		}
		
		return receivers;
	}
	
	/**
	 * An helper function which is used in order to clear the relevant information stored in the
	 * node. This is mainly used when there's a need to run the same network setup a few times without
	 * changing the neighbors each station has. (Requires only to re-send subscriptions)
	 */
	public void emptyGatheredInformationAndReworkRoutingTable()
	{
		// Clearing all the received messages
		this.receivedSubscriptionMessages.clear();
		this.receivedPublicationMessages.clear();
		
		// Clearing the routing table, because the radius may have been updated
		this.routingTable.clear();

		for (int index = 0 ; index < this.radius ; index++)
		{
			this.routingTable.add(new HashMap<Key, MyBloomFilter>());
			// Clearing each neighbors routing table
			for (Node neighbor : this.realNeighbours)
			{
				this.routingTable.get(index).put(neighbor.getKey(), new MyBloomFilter(Common.bloomFilterSize));
			}
		}
	}
}
